package job

import (
	"context"
	"fmt"
	"gcrontab/common"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
	"time"
)

type LogSink struct {
	client         *mongo.Client
	logCollection  *mongo.Collection
	logChan        chan *common.JobLog
	autoCommitChan chan *common.LogBath
}

var G_logSink *LogSink

func InitLogSink() (err error) {
	ctx, _ := context.WithTimeout(context.Background(), time.Duration(common.G_conifg.MongoTimeOut)*time.Millisecond)
	client, err := mongo.Connect(ctx, options.Client().ApplyURI(common.G_conifg.MongoUrl))
	if err != nil {
		return
	}

	G_logSink = &LogSink{
		client:         client,
		logCollection:  client.Database("cron").Collection("log"),
		logChan:        make(chan *common.JobLog, 1000),
		autoCommitChan: make(chan *common.LogBath, 1000),
	}

	//日志写入
	go G_logSink.writeLoop()

	return
}

//日志写入队列
func (l *LogSink) writeLoop() () {
	logBath := &common.LogBath{}
	for {
		select {
		case log := <-l.logChan:
			//先将日志写入队列
			logBath.Logs = append(logBath.Logs, log)

			//启动定时器
			//如果没有满100条，每隔10s，执行回调函数
			timer := time.AfterFunc(100*time.Second, func(bath *common.LogBath) (func()) {
				return func() {
					l.autoCommitChan <- bath //发送自动提交通知
				}
			}(logBath))

			//每100条，批量写入
			if len(logBath.Logs) >= 3 {
				l.saveLog(logBath)
				logBath = &common.LogBath{} //清空
				timer.Stop()                //清空定时器
			}
		case timeOutBath := <-l.autoCommitChan:
			if timeOutBath != logBath {
				continue
			}
			//写入mongodb
			l.saveLog(timeOutBath)
			logBath = &common.LogBath{}
		}
	}
}

//批量写入日志
func (l *LogSink) saveLog(bath *common.LogBath) {
	_, err := l.logCollection.InsertMany(context.TODO(), bath.Logs)
	if err != nil {
		fmt.Println("写入日志失败", err)
	}
}

//发送日志
func (l *LogSink) AppendLog(log *common.JobLog) () {
	select {
	case l.logChan <- log:
	default:
		//todo channel满了，日志丢弃，需改进
	}
}
